package com.ksyun.campus.dataserver.services;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.*;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.nio.charset.StandardCharsets;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.JsonNode;
import java.util.stream.Collectors;
import java.util.Base64;
import java.time.Duration;
import java.net.URLEncoder;
import jakarta.annotation.PostConstruct;

@Slf4j
@Service
public class DataService {

    @Value("${dataserver.data.path:/data}")
    private String dataPath;
    
    @Value("${dataserver.node-id:data-server-1}")
    private String nodeId;
    
    @Autowired
    private CuratorFramework zkClient;
    
    // 使用可配置的线程池大小
    @Value("${dataserver.replica.thread-pool-size:5}")
    private int replicaThreadPoolSize;
    
    // 使用可配置的超时时间
    @Value("${dataserver.replica.timeout-ms:30000}")
    private long replicaTimeoutMs;
    
    // 使用可配置的重试次数
    @Value("${dataserver.replica.max-retry-attempts:3}")
    private int maxRetryAttempts;
    
    private ExecutorService replicaExecutor;
    private HttpClient httpClient;
    private ObjectMapper objectMapper;
    
    // 三副本配置（可动态调整）
    private static final int DEFAULT_REQUIRED_REPLICAS = 3;
    
    /**
     * 根据当前可用DataServer数量动态计算所需副本数
     */
    private int getRequiredReplicas() {
        try {
            // 获取所有可用的DataServer
            List<String> availableServers = getAvailableDataServersSortedByUsage(new HashSet<>());
            // 包括本地节点
            int totalAvailable = availableServers.size() + 1;
            // 副本数不超过可用服务器数量，但至少为1
            return Math.min(DEFAULT_REQUIRED_REPLICAS, Math.max(1, totalAvailable));
        } catch (Exception e) {
            log.warn("Failed to get available servers for replica calculation, using default: {}", e.getMessage());
            return 1; // 默认情况下至少需要1个副本（本地）
        }
    }
    
    public DataService() {
    }

    @PostConstruct
    public void init() {
        int poolSize = replicaThreadPoolSize > 0 ? replicaThreadPoolSize : 5;
        this.replicaExecutor = Executors.newFixedThreadPool(poolSize);
        this.httpClient = HttpClient.newBuilder()
            .connectTimeout(Duration.ofSeconds(10))
            .build();
        this.objectMapper = new ObjectMapper();
    }
    
    /**
     * 写入文件数据，确保三副本
     * @deprecated 文件级存储已废弃，请使用块级存储API
     */
    @Deprecated
    public Map<String, Object> write(byte[] data, String fileSystemName, String path, int offset, int length) {
        Map<String, Object> result = new HashMap<>();
        result.put("success", false);
        result.put("message", "File-level write API is deprecated. Please use block-level write API: POST /api/v1/data/writeBlockWithData");
        result.put("errorCode", "DEPRECATED_API");
        result.put("suggestedApi", "POST /api/v1/data/writeBlockWithData");
        log.warn("File-level write API called but deprecated: fileSystemName={}, path={}", fileSystemName, path);
        return result;
    }

    /**
     * 副本节点本地写入（不触发进一步复制）
     * @deprecated 文件级存储已废弃，请使用块级存储API
     */
    @Deprecated
    public Map<String, Object> writeReplicaLocalOnly(byte[] data, String fileSystemName, String path, int offset, int length) {
        Map<String, Object> result = new HashMap<>();
        result.put("success", false);
        result.put("message", "File-level replica write API is deprecated. Please use block-level write API: POST /api/v1/data/writeBlockWithData");
        result.put("errorCode", "DEPRECATED_API");
        result.put("suggestedApi", "POST /api/v1/data/writeBlockWithData");
        log.warn("File-level replica write API called but deprecated: fileSystemName={}, path={}", fileSystemName, path);
        return result;
    }
    
    /**
     * 确保三副本写入，使用改进的重试机制
     */
    private boolean ensureThreeReplicas(byte[] data, String fileSystemName, String path, int offset, int length, List<String> replicaLocations) {
        int attempts = 0;
        List<String> availableDataServers;
        int requiredReplicas = getRequiredReplicas();
        
        log.info("Dynamic replica requirement: {} (based on available servers)", requiredReplicas);
        
        while (attempts < maxRetryAttempts && replicaLocations.size() < requiredReplicas) {
            attempts++;
            log.info("Replica attempt {}: Ensuring {} replicas, current count: {}", 
                    attempts, requiredReplicas, replicaLocations.size());
            
            // 获取可用的数据服务器（返回的是节点ID列表），按使用率升序以均衡分布
            Set<String> alreadyPlaced = replicaLocations.stream()
                    .map(s -> s.split(":", 2)[0])
                    .collect(Collectors.toSet());
            alreadyPlaced.add(nodeId);
            availableDataServers = getAvailableDataServersSortedByUsage(alreadyPlaced);
            if (availableDataServers.isEmpty()) {
                log.warn("No available data servers for replication on attempt {}", attempts);
                break;
            }
            
            // 计算需要写入的副本数量
            int neededReplicas = requiredReplicas - replicaLocations.size();
            int replicasToWrite = Math.min(neededReplicas, availableDataServers.size());
            
            log.info("Need {} more replicas, will write to {} servers", neededReplicas, replicasToWrite);
            
            // 并行写入副本
            List<CompletableFuture<ReplicaResult>> replicaFutures = new ArrayList<>();
            
            for (int i = 0; i < replicasToWrite; i++) {
                String dataServerNodeId = availableDataServers.get(i);
                CompletableFuture<ReplicaResult> future = CompletableFuture.supplyAsync(() -> {
                    return replicateToDataServer(dataServerNodeId, data, fileSystemName, path, offset, length);
                }, replicaExecutor);
                
                replicaFutures.add(future);
            }
            
            // 等待所有副本写入完成
            try {
                CompletableFuture.allOf(replicaFutures.toArray(new CompletableFuture[0]))
                    .get(replicaTimeoutMs, TimeUnit.MILLISECONDS);
                
                // 收集结果
                int successfulReplicas = 0;
                for (CompletableFuture<ReplicaResult> future : replicaFutures) {
                    try {
                        ReplicaResult replicaResult = future.get();
                        if (replicaResult.success) {
                            replicaLocations.add(replicaResult.dataServerId + ":" + replicaResult.path);
                            successfulReplicas++;
                        } else {
                            log.warn("Replica failed for server {}: {}", 
                                    replicaResult.dataServerId, replicaResult.errorMessage);
                        }
                    } catch (Exception e) {
                        log.warn("Failed to get replica result: {}", e.getMessage());
                    }
                }
                
                log.info("Replica write attempt {}: {} successful out of {} attempted", 
                        attempts, successfulReplicas, replicasToWrite);
                
            } catch (Exception e) {
                log.warn("Replica write attempt {} failed: {}", attempts, e.getMessage());
            }
        }
        
        boolean success = replicaLocations.size() >= requiredReplicas;
        log.info("Final replica count: {} (required: {}), success: {}", 
                replicaLocations.size(), requiredReplicas, success);
        
        return success;
    }
    
    /**
     * 获取可用的数据服务器列表，改进错误处理
     */
    private List<String> getAvailableDataServersSortedByUsage(Set<String> excludeNodeIds) {
        class NodeUsage {
            final String id; final double usage; NodeUsage(String id, double usage){ this.id=id; this.usage=usage; }
        }
        List<NodeUsage> nodes = new ArrayList<>();
        try {
            String base = "/minfs/dataservers";
            List<String> children = zkClient.getChildren().forPath(base);
            for (String child : children) {
                if ("heartbeat".equals(child)) continue;
                if (excludeNodeIds != null && excludeNodeIds.contains(child)) continue;
                double usage = Double.MAX_VALUE;
                try {
                    String hbPath = base + "/heartbeat/" + child;
                    if (zkClient.checkExists().forPath(hbPath) != null) {
                        byte[] hb = zkClient.getData().forPath(hbPath);
                        String json = new String(hb, StandardCharsets.UTF_8);
                        JsonNode j = objectMapper.readTree(json);
                        if (j.has("capacityBytes") && j.has("usedBytes")) {
                            long cap = Math.max(1L, j.get("capacityBytes").asLong(1));
                            long used = j.get("usedBytes").asLong(0);
                            usage = (double) used / (double) cap;
                        } else if (j.has("usagePercent")) {
                            usage = j.get("usagePercent").asDouble(100.0) / 100.0;
                        } else {
                            usage = 1.0; // 无信息时放末尾
                        }
                    }
                } catch (Exception ignore) {
                    usage = 1.0;
                }
                nodes.add(new NodeUsage(child, usage));
            }
            nodes.sort(Comparator.comparingDouble(n -> n.usage));
            return nodes.stream().map(n -> n.id).collect(Collectors.toList());
        } catch (Exception e) {
            log.error("Error getting available data servers by usage", e);
            return new ArrayList<>();
        }
    }
    
    /**
     * 复制数据到其他数据服务器，改进错误处理
     */
    private ReplicaResult replicateToDataServer(String dataServerId, byte[] data, String fileSystemName, String path, int offset, int length) {
        try {
            // 从ZooKeeper获取数据服务器信息
            String dataServerPath = "/minfs/dataservers/" + dataServerId;
            byte[] nodeData = zkClient.getData().forPath(dataServerPath);
            String nodeInfoStr = new String(nodeData, StandardCharsets.UTF_8);
            
            // 解析节点信息获取 host 与 端口
            String host = "localhost";
            int port = getPortFromNodeId(dataServerId);  // 先用nodeId推断端口
            try {
                JsonNode jsonNode = objectMapper.readTree(nodeInfoStr);
                if (jsonNode.has("host")) {
                    host = jsonNode.get("host").asText("localhost");
                }
                if (jsonNode.has("port")) {
                    port = jsonNode.get("port").asInt(port);  // 使用推断的端口作为默认值
                }
            } catch (Exception parseEx) {
                // 解析失败时，回退到基于 nodeId 的端口推断
                port = getPortFromNodeId(dataServerId);
            }
            
            // 调用远程数据服务器的写入接口
            String url = String.format("http://%s:%d/api/write", host, port);
            
            // 添加查询参数
            String queryParams = String.format("?path=%s&offset=%d&length=%d", 
                    URLEncoder.encode(path, java.nio.charset.StandardCharsets.UTF_8), offset, length);
            String fullUrl = url + queryParams;
            
            // 构建HTTP请求
            HttpRequest request = HttpRequest.newBuilder()
                    .uri(URI.create(fullUrl))
                    .header("fileSystemName", fileSystemName)
                    .header("Content-Type", "application/octet-stream")
                    .header("X-Source-Node", nodeId)
                    .POST(HttpRequest.BodyPublishers.ofByteArray(data))
                    .timeout(java.time.Duration.ofSeconds(10))
                    .build();
            
            // 发送HTTP请求
            HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
            
            if (response.statusCode() == 200) {
                log.info("Successfully replicated to data server: {}:{}", dataServerId, port);
                return new ReplicaResult(true, dataServerId, dataServerId + ":" + path, null);
            } else {
                log.error("Failed to replicate to data server: {}:{}, status: {}", 
                        dataServerId, port, response.statusCode());
                return new ReplicaResult(false, dataServerId, null, "HTTP status: " + response.statusCode());
            }
            
        } catch (Exception e) {
            log.error("Error replicating to data server: {}", dataServerId, e);
            return new ReplicaResult(false, dataServerId, null, e.getMessage());
        }
    }
    
    /**
     * 清理本地文件，改进错误处理
     */
    private void cleanupLocalFile(String fileSystemName, String path) {
        try {
            Path filePath = resolveSafePath(fileSystemName, path);
            if (Files.exists(filePath)) {
                Files.delete(filePath);
                log.info("Cleaned up local file after replication failure: {}", path);
            }
        } catch (Exception e) {
            log.error("Failed to cleanup local file: {}", path, e);
        }
    }

    /**
     * 仅在本次写入前文件不存在时才清理，避免误删历史文件
     */
    private void cleanupLocalFileIfNew(String fileSystemName, String path, boolean existedBefore) {
        if (!existedBefore) {
            cleanupLocalFile(fileSystemName, path);
        }
    }
    
    /**
     * 写入本地文件，改进错误处理
     */
    private String writeToLocal(byte[] data, String fileSystemName, String path, int offset, int length) throws IOException {
        Path filePath = resolveSafePath(fileSystemName, path);
        
        // 确保父目录存在
        Path parentDir = filePath.getParent();
        if (parentDir != null && !Files.exists(parentDir)) {
            Files.createDirectories(parentDir);
        }
        
        try (RandomAccessFile raf = new RandomAccessFile(filePath.toFile(), "rw");
             java.nio.channels.FileChannel channel = raf.getChannel();
             java.nio.channels.FileLock lock = channel.lock()) {
            raf.seek(offset);
            raf.write(data, 0, Math.min(data.length, length));
        }
        
        return filePath.toString();
    }
    
    /**
     * 从节点ID获取端口，改进端口映射逻辑
     */
    private int getPortFromNodeId(String nodeId) {
        try {
            // 从nodeId中提取索引
            String indexStr = nodeId.replaceAll("data-server-", "");
            int index = Integer.parseInt(indexStr);
            
            // 端口映射：9001, 9002, 9003, 9004 (与启动脚本保持一致)
            switch (index) {
                case 1: return 9001;
                case 2: return 9002;
                case 3: return 9003;
                case 4: return 9004;
                default: return 9001 + (index - 1);
            }
        } catch (Exception e) {
            log.warn("Could not parse port from nodeId: {}, using default: 9001", nodeId);
            return 9001;
        }
    }
    
    /**
     * 读取文件数据，改进错误处理
     * @deprecated 文件级存储已废弃，请使用块级存储API
     */
    @Deprecated
    public byte[] read(String fileSystemName, String path, int offset, int length) {
        log.warn("File-level read API called but deprecated: fileSystemName={}, path={}, offset={}, length={}", 
                fileSystemName, path, offset, length);
        return null; // 返回null表示文件不存在，符合原有API语义
    }
    
    /**
     * 删除文件，改进错误处理
     * @deprecated 文件级存储已废弃，请使用块级存储API
     */
    @Deprecated
    public boolean delete(String fileSystemName, String path) {
        log.warn("File-level delete API called but deprecated: fileSystemName={}, path={}", fileSystemName, path);
        return false; // 返回false表示删除失败，符合原有API语义
    }

    /**
     * 目录递归删除/非递归删除
     */
    public Map<String, Object> deleteDirectory(String fileSystemName, String path, boolean recursive) {
        Map<String, Object> result = new HashMap<>();
        Path dirPath = resolveSafePath(fileSystemName, path);
        try {
            if (!Files.exists(dirPath)) {
                result.put("success", false);
                result.put("errorCode", "NOT_FOUND");
                result.put("message", "Directory not found");
                return result;
            }
            if (!Files.isDirectory(dirPath)) {
                result.put("success", false);
                result.put("errorCode", "NOT_DIRECTORY");
                result.put("message", "Path is not a directory");
                return result;
            }
            if (!recursive) {
                // 非递归：只允许空目录
                try (java.util.stream.Stream<Path> s = Files.list(dirPath)) {
                    if (s.findAny().isPresent()) {
                        result.put("success", false);
                        result.put("errorCode", "NOT_EMPTY");
                        result.put("message", "Directory is not empty");
                        return result;
                    }
                }
                Files.delete(dirPath);
            } else {
                // 递归删除
                Files.walk(dirPath)
                        .sorted(java.util.Comparator.reverseOrder())
                        .forEach(p -> {
                            try { Files.delete(p); } catch (Exception ignore) {}
                        });
            }
            result.put("success", true);
            result.put("message", "Directory deleted");
            return result;
        } catch (Exception e) {
            log.error("Error deleting directory: {}", path, e);
            result.put("success", false);
            result.put("errorCode", "ERROR");
            result.put("message", e.getMessage());
            return result;
        }
    }
    
    /**
     * 获取文件信息，改进错误处理
     */
    public Map<String, Object> getFileInfo(String fileSystemName, String path) {
        Map<String, Object> fileInfo = new HashMap<>();
        try {
            Path filePath = resolveSafePath(fileSystemName, path);
            if (Files.exists(filePath)) {
                fileInfo.put("exists", true);
                fileInfo.put("path", path);
                fileInfo.put("size", Files.size(filePath));
                fileInfo.put("lastModified", Files.getLastModifiedTime(filePath).toMillis());
                fileInfo.put("isDirectory", Files.isDirectory(filePath));
                fileInfo.put("isReadable", Files.isReadable(filePath));
                fileInfo.put("isWritable", Files.isWritable(filePath));
            } else {
                fileInfo.put("exists", false);
            }
        } catch (Exception e) {
            log.error("Error getting file info: {}", path, e);
            fileInfo.put("error", e.getMessage());
        }
        return fileInfo;
    }

    /**
     * 目录列表
     */
    public Map<String, Object> listDirectory(String fileSystemName, String path) {
        Map<String, Object> result = new HashMap<>();
        Path dirPath = resolveSafePath(fileSystemName, path);
        try {
            if (!Files.exists(dirPath)) {
                result.put("success", false);
                result.put("errorCode", "NOT_FOUND");
                result.put("message", "Path not found");
                return result;
            }
            if (!Files.isDirectory(dirPath)) {
                result.put("success", false);
                result.put("errorCode", "NOT_DIRECTORY");
                result.put("message", "Path is not a directory");
                return result;
            }
            java.util.List<Map<String, Object>> entries = new java.util.ArrayList<>();
            try (java.util.stream.Stream<Path> stream = Files.list(dirPath)) {
                stream.forEach(p -> {
                    Map<String, Object> e = new HashMap<>();
                    e.put("name", p.getFileName().toString());
                    e.put("isDirectory", Files.isDirectory(p));
                    try { e.put("size", Files.isDirectory(p) ? 0 : Files.size(p)); } catch (Exception ignore) {}
                    try { e.put("lastModified", Files.getLastModifiedTime(p).toMillis()); } catch (Exception ignore) {}
                    entries.add(e);
                });
            }
            result.put("success", true);
            result.put("path", path);
            result.put("entries", entries);
            result.put("count", entries.size());
            return result;
        } catch (Exception e) {
            log.error("Error listing directory: {}", path, e);
            result.put("success", false);
            result.put("errorCode", "ERROR");
            result.put("message", e.getMessage());
            return result;
        }
    }
    
    /**
     * 获取副本信息，改进错误处理
     */
    public Map<String, Object> getReplicaInfo(String fileSystemName, String path) {
        Map<String, Object> replicaInfo = new HashMap<>();
        List<Map<String, Object>> replicas = new ArrayList<>();
        
        try {
            // 获取所有数据服务器
            String dataServerPath = "/minfs/dataservers";
            List<String> children = zkClient.getChildren().forPath(dataServerPath);
            
            for (String child : children) {
                if (!child.startsWith("heartbeat")) {
                    Map<String, Object> replica = new HashMap<>();
                    replica.put("nodeId", child);
                    
                    // 检查本地文件是否存在
                    Path localPath = Paths.get(dataPath, fileSystemName, path);
                    if (Files.exists(localPath)) {
                        replica.put("status", "available");
                        replica.put("size", Files.size(localPath));
                        replica.put("lastModified", Files.getLastModifiedTime(localPath).toMillis());
                    } else {
                        replica.put("status", "missing");
                    }
                    
                    replicas.add(replica);
                }
            }
            
            replicaInfo.put("filePath", path);
            replicaInfo.put("replicas", replicas);
            replicaInfo.put("totalReplicas", replicas.size());
            replicaInfo.put("availableReplicas", replicas.stream()
                    .filter(r -> "available".equals(r.get("status")))
                    .count());
            
        } catch (Exception e) {
            log.error("Error getting replica info: {}", path, e);
            replicaInfo.put("error", e.getMessage());
        }
        
        return replicaInfo;
    }

    /**
     * 获取跨节点副本分布（读取 ZK 节点并远程 HEAD 探测）
     */
    public Map<String, Object> getReplicaMap(String fileSystemName, String path) {
        Map<String, Object> data = new HashMap<>();
        List<Map<String, Object>> nodes = new ArrayList<>();
        try {
            String base = "/minfs/dataservers";
            List<String> children = zkClient.getChildren().forPath(base);
            for (String child : children) {
                if ("heartbeat".equals(child)) continue;
                Map<String, Object> n = new HashMap<>();
                n.put("nodeId", child);
                try {
                    byte[] nodeData = zkClient.getData().forPath(base + "/" + child);
                    String json = new String(nodeData, StandardCharsets.UTF_8);
                    String host = "localhost"; int port = getPortFromNodeId(nodeId);
                    try {
                        JsonNode j = objectMapper.readTree(json);
                        if (j.has("host")) host = j.get("host").asText("localhost");
                        if (j.has("port")) port = j.get("port").asInt(port);
                    } catch (Exception ignore) {}
                    // 通过文件信息接口判断存在性
                    String url = String.format("http://%s:%d/api/info?path=%s", host, port, path);
                    HttpRequest req = HttpRequest.newBuilder()
                            .uri(URI.create(url))
                            .header("fileSystemName", fileSystemName)
                            .timeout(java.time.Duration.ofSeconds(3))
                            .GET()
                            .build();
                    HttpResponse<String> resp = httpClient.send(req, HttpResponse.BodyHandlers.ofString());
                    boolean exists = resp.statusCode() == 200 && resp.body() != null && resp.body().contains("\"exists\":true");
                    n.put("exists", exists);
                } catch (Exception ex) {
                    n.put("exists", false);
                    n.put("error", ex.getMessage());
                }
                nodes.add(n);
            }
            long count = nodes.stream().filter(m -> Boolean.TRUE.equals(m.get("exists"))).count();
            data.put("filePath", path);
            data.put("replicas", nodes);
            data.put("replicaCount", count);
            try {
                Map<String, Object> meta = readFileReplicaMetadata(fileSystemName, path);
                if (meta != null) {
                    data.put("metadata", meta);
                }
            } catch (Exception ignore) {}
            return data;
        } catch (Exception e) {
            data.put("error", e.getMessage());
            return data;
        }
    }

    // ======= 副本元数据（ZK）与分布统计 =======
    private String buildFileKey(String fileSystemName, String path) {
        String raw = fileSystemName + ":" + path;
        return Base64.getUrlEncoder().withoutPadding().encodeToString(raw.getBytes(StandardCharsets.UTF_8));
    }
    private String buildFileMetaPath(String fileSystemName, String path) {
        return "/minfs/files/" + buildFileKey(fileSystemName, path);
    }
    private byte[] toJsonBytes(Map<String, Object> map) throws Exception {
        return objectMapper.writeValueAsBytes(map);
    }
    private Map<String, Object> fromJsonBytes(byte[] bytes) throws Exception {
        return objectMapper.readValue(bytes, new com.fasterxml.jackson.core.type.TypeReference<Map<String, Object>>() {});
    }
    public void updateFileReplicaMetadata(String fileSystemName, String path, List<String> replicas, String primary) {
        try {
            String metaPath = buildFileMetaPath(fileSystemName, path);
            Map<String, Object> meta = new HashMap<>();
            meta.put("fileSystemName", fileSystemName);
            meta.put("path", path);
            meta.put("primary", primary);
            meta.put("replicas", new ArrayList<>(new LinkedHashSet<>(replicas)));
            meta.put("updatedAt", System.currentTimeMillis());
            byte[] data = toJsonBytes(meta);
            if (zkClient.checkExists().forPath(metaPath) == null) {
                zkClient.create().creatingParentsIfNeeded().forPath(metaPath, data);
            } else {
                zkClient.setData().forPath(metaPath, data);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    public void removeReplicaFromMetadata(String fileSystemName, String path, String replicaNodeId) {
        try {
            String metaPath = buildFileMetaPath(fileSystemName, path);
            if (zkClient.checkExists().forPath(metaPath) == null) return;
            Map<String, Object> meta = fromJsonBytes(zkClient.getData().forPath(metaPath));
            List<?> replicasAny = (List<?>) meta.getOrDefault("replicas", new ArrayList<>());
            List<String> replicas = replicasAny.stream().map(Object::toString).collect(Collectors.toList());
            replicas = replicas.stream().filter(id -> !Objects.equals(id, replicaNodeId)).collect(Collectors.toList());
            if (replicas.isEmpty()) {
                zkClient.delete().forPath(metaPath);
            } else {
                meta.put("replicas", replicas);
                if (Objects.equals(meta.get("primary"), replicaNodeId)) {
                    meta.put("primary", replicas.get(0));
                }
                meta.put("updatedAt", System.currentTimeMillis());
                zkClient.setData().forPath(metaPath, toJsonBytes(meta));
            }
        } catch (Exception e) {
            log.warn("removeReplicaFromMetadata failed for {}:{} - {}", fileSystemName, path, e.getMessage());
        }
    }
    public Map<String, Object> readFileReplicaMetadata(String fileSystemName, String path) {
        try {
            String metaPath = buildFileMetaPath(fileSystemName, path);
            if (zkClient.checkExists().forPath(metaPath) == null) return null;
            return fromJsonBytes(zkClient.getData().forPath(metaPath));
        } catch (Exception e) {
            return null;
        }
    }
    public Map<String, Object> computeDistributionStats() {
        Map<String, Object> result = new HashMap<>();
        Map<String, Integer> primaryPerNode = new HashMap<>();
        Map<String, Integer> replicasPerNode = new HashMap<>();
        try {
            String base = "/minfs/files";
            if (zkClient.checkExists().forPath(base) == null) {
                result.put("primaryPerNode", primaryPerNode);
                result.put("replicasPerNode", replicasPerNode);
                return result;
            }
            List<String> children = zkClient.getChildren().forPath(base);
            for (String ch : children) {
                byte[] data = zkClient.getData().forPath(base + "/" + ch);
                Map<String, Object> meta = fromJsonBytes(data);
                Object p = meta.get("primary");
                if (p != null) {
                    primaryPerNode.put(p.toString(), primaryPerNode.getOrDefault(p.toString(), 0) + 1);
                }
                List<?> repsAny = (List<?>) meta.getOrDefault("replicas", new ArrayList<>());
                for (Object r : repsAny) {
                    String id = String.valueOf(r);
                    replicasPerNode.put(id, replicasPerNode.getOrDefault(id, 0) + 1);
                }
            }
        } catch (Exception e) {
            result.put("error", e.getMessage());
        }
        result.put("primaryPerNode", primaryPerNode);
        result.put("replicasPerNode", replicasPerNode);
        result.put("timestamp", System.currentTimeMillis());
        return result;
    }

    // ======= FSCK/Repair 支持 =======
    private static final int DEFAULT_MAX_READ_CHUNK = 128 * 1024 * 1024; // 128MB

    /**
     * 复制字节到指定节点
     * @deprecated 文件级存储已废弃，请使用块级存储API
     */
    @Deprecated
    public boolean replicateBytesToNode(String targetNodeId, byte[] data, String fileSystemName, String path, int offset, int length) {
        log.warn("File-level replicateBytesToNode API called but deprecated: targetNodeId={}, fileSystemName={}, path={}", 
                targetNodeId, fileSystemName, path);
        return false; // 返回false表示复制失败，符合原有API语义
    }

    // ====== 块级接口实现（与 MetaServer 对齐）======
    
    /**
     * 块级写入 - 主要写入接口
     * 支持主副本写入（触发副本复制）和从副本写入（仅本地存储）
     */
    public Map<String, Object> writeBlock(String fileId, int blkIdx, byte[] data, boolean isPrimary, String sourceNodeId) {
        Map<String, Object> result = new HashMap<>();
        
        if (data == null || data.length == 0) {
            result.put("success", false);
            result.put("message", "Block data cannot be null or empty");
            result.put("errorCode", "INVALID_DATA");
            return result;
        }
        
        if (fileId == null || fileId.trim().isEmpty()) {
            result.put("success", false);
            result.put("message", "FileId is required");
            result.put("errorCode", "INVALID_FILEID");
            return result;
        }
        
        if (blkIdx < 0) {
            result.put("success", false);
            result.put("message", "Block index must be non-negative");
            result.put("errorCode", "INVALID_BLKIDX");
            return result;
        }
        
        try {
            // 1. 写入本地块存储
            Path blockPath = writeBlockLocal(fileId, blkIdx, data);
            log.info("Successfully wrote block locally: fileId={}, blkIdx={}, size={}, path={}", 
                    fileId, blkIdx, data.length, blockPath);
            
            // 2. 如果是主副本写入，触发副本复制
            List<String> replicaLocations = new ArrayList<>();
            replicaLocations.add(nodeId + ":" + blockPath.toString());
            
            if (isPrimary) {
                boolean replicaSuccess = ensureBlockReplicas(fileId, blkIdx, data, replicaLocations);
                if (!replicaSuccess) {
                    // 副本复制失败，但本地已写入成功，记录警告
                    log.warn("Block replicas incomplete for fileId={}, blkIdx={}, only {} replicas", 
                            fileId, blkIdx, replicaLocations.size());
                }
            }
            
            // 3. 构造成功响应
            result.put("success", true);
            result.put("message", "Block written successfully");
            result.put("fileId", fileId);
            result.put("blkIdx", blkIdx);
            result.put("size", data.length);
            result.put("blockPath", blockPath.toString());
            result.put("replicaCount", replicaLocations.size());
            result.put("replicaLocations", replicaLocations);
            result.put("isPrimary", isPrimary);
            result.put("timestamp", System.currentTimeMillis());
            
            return result;
            
        } catch (Exception e) {
            log.error("Error writing block: fileId={}, blkIdx={}", fileId, blkIdx, e);
            result.put("success", false);
            result.put("message", "Failed to write block: " + e.getMessage());
            result.put("errorCode", "WRITE_ERROR");
            result.put("exception", e.getClass().getSimpleName());
            return result;
        }
    }
    
    /**
     * 确保块的副本复制
     */
    private boolean ensureBlockReplicas(String fileId, int blkIdx, byte[] data, List<String> replicaLocations) {
        int requiredReplicas = getRequiredReplicas();
        int attempts = 0;
        
        log.info("Ensuring block replicas: fileId={}, blkIdx={}, required={}, current={}", 
                fileId, blkIdx, requiredReplicas, replicaLocations.size());
        
        while (attempts < maxRetryAttempts && replicaLocations.size() < requiredReplicas) {
            attempts++;
            
            // 获取可用的数据服务器
            Set<String> alreadyPlaced = replicaLocations.stream()
                    .map(s -> s.split(":", 2)[0])
                    .collect(Collectors.toSet());
            alreadyPlaced.add(nodeId);
            
            List<String> availableDataServers = getAvailableDataServersSortedByUsage(alreadyPlaced);
            if (availableDataServers.isEmpty()) {
                log.warn("No available data servers for block replication on attempt {}", attempts);
                break;
            }
            
            // 计算需要写入的副本数量
            int neededReplicas = requiredReplicas - replicaLocations.size();
            int replicasToWrite = Math.min(neededReplicas, availableDataServers.size());
            
            log.info("Block replica attempt {}: need {} more replicas, will write to {} servers", 
                    attempts, neededReplicas, replicasToWrite);
            
            // 并行写入副本
            List<CompletableFuture<BlockReplicaResult>> replicaFutures = new ArrayList<>();
            
            for (int i = 0; i < replicasToWrite; i++) {
                String dataServerNodeId = availableDataServers.get(i);
                CompletableFuture<BlockReplicaResult> future = CompletableFuture.supplyAsync(() -> {
                    return replicateBlockToDataServer(dataServerNodeId, fileId, blkIdx, data);
                }, replicaExecutor);
                
                replicaFutures.add(future);
            }
            
            // 等待所有副本写入完成
            try {
                CompletableFuture.allOf(replicaFutures.toArray(new CompletableFuture[0]))
                    .get(replicaTimeoutMs, TimeUnit.MILLISECONDS);
                
                // 收集结果
                int successfulReplicas = 0;
                for (CompletableFuture<BlockReplicaResult> future : replicaFutures) {
                    try {
                        BlockReplicaResult replicaResult = future.get();
                        if (replicaResult.success) {
                            replicaLocations.add(replicaResult.dataServerId + ":" + replicaResult.blockPath);
                            successfulReplicas++;
                        } else {
                            log.warn("Block replica failed for server {}: {}", 
                                    replicaResult.dataServerId, replicaResult.errorMessage);
                        }
                    } catch (Exception e) {
                        log.warn("Failed to get block replica result: {}", e.getMessage());
                    }
                }
                
                log.info("Block replica attempt {}: {} successful out of {} attempted", 
                        attempts, successfulReplicas, replicasToWrite);
                
            } catch (Exception e) {
                log.warn("Block replica attempt {} failed: {}", attempts, e.getMessage());
            }
        }
        
        boolean success = replicaLocations.size() >= requiredReplicas;
        log.info("Final block replica count: {} (required: {}), success: {}", 
                replicaLocations.size(), requiredReplicas, success);
        
        return success;
    }
    
    /**
     * 复制块到其他数据服务器
     */
    private BlockReplicaResult replicateBlockToDataServer(String dataServerId, String fileId, int blkIdx, byte[] data) {
        try {
            // 从ZooKeeper获取数据服务器信息
            String dataServerPath = "/minfs/dataservers/" + dataServerId;
            byte[] nodeData = zkClient.getData().forPath(dataServerPath);
            String nodeInfoStr = new String(nodeData, StandardCharsets.UTF_8);
            
            // 解析节点信息获取 host 与 端口
            String host = "localhost";
            int port = getPortFromNodeId(dataServerId);
            try {
                JsonNode jsonNode = objectMapper.readTree(nodeInfoStr);
                if (jsonNode.has("host")) {
                    host = jsonNode.get("host").asText("localhost");
                }
                if (jsonNode.has("port")) {
                    port = jsonNode.get("port").asInt(port);
                }
            } catch (Exception parseEx) {
                port = getPortFromNodeId(dataServerId);
            }
            
            // 调用远程数据服务器的块写入接口（携带二进制数据）
            String url = String.format(
                    "http://%s:%d/api/v1/data/writeBlockWithData?fileId=%s&blkIdx=%d&isPrimary=%s&sourceNodeId=%s",
                    host, port,
                    URLEncoder.encode(fileId, StandardCharsets.UTF_8),
                    blkIdx,
                    "false",
                    URLEncoder.encode(nodeId, StandardCharsets.UTF_8)
            );

            // 构建HTTP请求（octet-stream）
            HttpRequest request = HttpRequest.newBuilder()
                    .uri(URI.create(url))
                    .header("Content-Type", "application/octet-stream")
                    .header("X-Source-Node", nodeId)
                    .POST(HttpRequest.BodyPublishers.ofByteArray(data))
                    .timeout(Duration.ofSeconds(10))
                    .build();
            
            // 发送HTTP请求
            HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
            
            if (response.statusCode() == 200) {
                // 解析响应获取块路径
                Map<String, Object> responseData = objectMapper.readValue(response.body(),
                        new com.fasterxml.jackson.core.type.TypeReference<Map<String, Object>>() {});

                if (Boolean.TRUE.equals(responseData.get("success"))) {
                    String blockPath = (String) responseData.get("blockPath");
                    log.info("Successfully replicated block to data server: {}:{}, fileId={}, blkIdx={}",
                            dataServerId, port, fileId, blkIdx);
                    return new BlockReplicaResult(true, dataServerId, blockPath, null);
                } else {
                    String errorMsg = String.valueOf(responseData.getOrDefault("message", "unknown error"));
                    log.error("Block replication failed on remote server {}:{}: {}",
                            dataServerId, port, errorMsg);
                    return new BlockReplicaResult(false, dataServerId, null, errorMsg);
                }
            } else {
                log.error("Failed to replicate block to data server: {}:{}, status: {}", 
                        dataServerId, port, response.statusCode());
                return new BlockReplicaResult(false, dataServerId, null, "HTTP status: " + response.statusCode());
            }
            
        } catch (Exception e) {
            log.error("Error replicating block to data server: {}", dataServerId, e);
            return new BlockReplicaResult(false, dataServerId, null, e.getMessage());
        }
    }
    
    /**
     * 块副本结果内部类
     */
    private static class BlockReplicaResult {
        final boolean success;
        final String dataServerId;
        final String blockPath;
        final String errorMessage;
        
        BlockReplicaResult(boolean success, String dataServerId, String blockPath, String errorMessage) {
            this.success = success;
            this.dataServerId = dataServerId;
            this.blockPath = blockPath;
            this.errorMessage = errorMessage;
        }
    }

    public Map<String, Object> replicateBlockFromRemote(String srcHost, int srcPort, String fileId, int blkIdx, long size) {
        Map<String, Object> res = new HashMap<>();
        try {
            byte[] content = fetchBlockFromRemote(srcHost, srcPort, fileId, blkIdx, size);
            if (content == null) {
                res.put("success", false);
                res.put("message", "source block not available");
                return res;
            }
            
            // 写入到标准块存储位置
            Path path = writeBlockLocal(fileId, blkIdx, content);
            
            // 同时尝试推断并更新文件级存储（如果能找到对应的文件路径）
            try {
                updateFileStorageIfNeeded(fileId, blkIdx, content);
            } catch (Exception e) {
                log.debug("Failed to update file storage for {}:{}", fileId, blkIdx, e);
                // 不影响主要的块复制流程
            }
            
            res.put("success", true);
            res.put("bytes", content.length);
            res.put("path", path.toString());
            return res;
        } catch (Exception e) {
            log.error("replicateBlockFromRemote error", e);
            res.put("success", false);
            res.put("message", e.getMessage());
            return res;
        }
    }
    
    /**
     * 如果能找到对应的文件路径，同时更新文件级存储
     */
    private void updateFileStorageIfNeeded(String fileId, int blkIdx, byte[] content) {
        try {
            // 查找对应的文件路径
            String metaPath = "/minfs/files";
            if (zkClient.checkExists().forPath(metaPath) == null) {
                return;
            }
            
            List<String> children = zkClient.getChildren().forPath(metaPath);
            for (String childPath : children) {
                try {
                    byte[] data = zkClient.getData().forPath(metaPath + "/" + childPath);
                    Map<String, Object> meta = fromJsonBytes(data);
                    
                    String metaFileSystem = (String) meta.get("fileSystemName");
                    String filePath = (String) meta.get("path");
                    
                    if (metaFileSystem != null && filePath != null) {
                        if (isFileMatchingId(metaFileSystem, filePath, fileId)) {
                            // 找到对应的文件，更新它
                            updateFileWithBlock(metaFileSystem, filePath, blkIdx, content);
                            log.debug("Updated file storage for {}:{} -> {}:{}", 
                                    fileId, blkIdx, metaFileSystem, filePath);
                            return;
                        }
                    }
                } catch (Exception ignore) {
                    // 继续检查其他文件
                }
            }
        } catch (Exception e) {
            log.debug("Error updating file storage for {}:{}", fileId, blkIdx, e);
        }
    }
    
    /**
     * 使用块数据更新文件
     */
    private void updateFileWithBlock(String fileSystemName, String filePath, int blkIdx, byte[] blockContent) {
        try {
            Path actualFile = resolveSafePath(fileSystemName, filePath);
            
            // 确保父目录存在
            Path parentDir = actualFile.getParent();
            if (parentDir != null && !Files.exists(parentDir)) {
                Files.createDirectories(parentDir);
            }
            
            // 如果是第0块且文件不存在，直接写入
            if (blkIdx == 0 && !Files.exists(actualFile)) {
                Files.write(actualFile, blockContent);
                log.debug("Created file {} with block 0", actualFile);
                return;
            }
            
            // 对于其他情况，需要在正确的偏移位置写入
            int blockSize = 524288; // 512KB
            long offset = (long) blkIdx * blockSize;
            
            try (RandomAccessFile raf = new RandomAccessFile(actualFile.toFile(), "rw")) {
                raf.seek(offset);
                raf.write(blockContent);
                log.debug("Updated file {} block {} at offset {}", actualFile, blkIdx, offset);
            }
            
        } catch (Exception e) {
            log.warn("Failed to update file {}:{} with block {}", fileSystemName, filePath, blkIdx, e);
        }
    }

    public Map<String, Object> deleteBlock(String fileId, int blkIdx) {
        Map<String, Object> res = new HashMap<>();
        try {
            Path p = getBlockPath(fileId, blkIdx);
            boolean existed = Files.exists(p);
            if (existed) {
                Files.delete(p);
            }
            res.put("success", true);
            res.put("existed", existed);
            return res;
        } catch (Exception e) {
            log.error("deleteBlock error", e);
            res.put("success", false);
            res.put("message", e.getMessage());
            return res;
        }
    }

    /**
     * 删除文件的所有块
     * 根据fileId删除该文件的所有块文件
     */
    public Map<String, Object> deleteFileBlocks(String fileId) {
        Map<String, Object> res = new HashMap<>();
        try {
            String safeId = sanitizeFileId(fileId);
            Path blocksDir = Paths.get(dataPath).resolve("blocks").resolve(safeId);
            
            if (!Files.exists(blocksDir)) {
                res.put("success", true);
                res.put("message", "File blocks directory does not exist");
                res.put("deletedBlocks", 0);
                return res;
            }
            
            // 删除该文件的所有块文件
            int deletedCount = 0;
            try (var stream = Files.walk(blocksDir)) {
                var files = stream.filter(Files::isRegularFile)
                                 .filter(p -> p.getFileName().toString().startsWith("blk_"))
                                 .collect(Collectors.toList());
                
                for (Path file : files) {
                    try {
                        Files.delete(file);
                        deletedCount++;
                    } catch (IOException e) {
                        log.warn("Failed to delete block file: {}", file, e);
                    }
                }
            }
            
            // 尝试删除空的目录
            try {
                if (Files.exists(blocksDir) && Files.isDirectory(blocksDir)) {
                    Files.delete(blocksDir);
                }
            } catch (IOException e) {
                log.warn("Failed to delete empty blocks directory: {}", blocksDir, e);
            }
            
            res.put("success", true);
            res.put("message", "Successfully deleted " + deletedCount + " blocks");
            res.put("deletedBlocks", deletedCount);
            res.put("fileId", fileId);
            
            log.info("Deleted {} blocks for fileId: {}", deletedCount, fileId);
            return res;
            
        } catch (Exception e) {
            log.error("deleteFileBlocks error for fileId: {}", fileId, e);
            res.put("success", false);
            res.put("message", e.getMessage());
            res.put("fileId", fileId);
            return res;
        }
    }

    public byte[] readBlock(String fileId, int blkIdx) {
        try {
            // 首先尝试从标准块存储路径读取
            Path blockPath = getBlockPath(fileId, blkIdx);
            if (Files.exists(blockPath)) {
                return Files.readAllBytes(blockPath);
            }
            
            // 如果块存储中不存在，尝试从文件级存储中读取
            return readBlockFromFileStorage(fileId, blkIdx);
            
        } catch (Exception e) {
            log.error("readBlock error for {}:{}", fileId, blkIdx, e);
            return null;
        }
    }
    
    /**
     * 从文件级存储中读取块数据
     */
    private byte[] readBlockFromFileStorage(String fileId, int blkIdx) {
        try {
            // 查找对应的文件
            Path dataDir = Paths.get(dataPath);
            if (!Files.exists(dataDir)) {
                return null;
            }
            
            try (java.util.stream.Stream<Path> fsDirs = Files.list(dataDir)) {
                return fsDirs.filter(Files::isDirectory)
                        .map(fsDir -> {
                            String fsName = fsDir.getFileName().toString();
                            if ("blocks".equals(fsName)) return null; // 跳过块存储目录
                            
                            return readBlockFromFileSystem(fsName, fileId, blkIdx);
                        })
                        .filter(Objects::nonNull)
                        .findFirst()
                        .orElse(null);
            }
            
        } catch (Exception e) {
            log.debug("Error reading block from file storage for {}:{}", fileId, blkIdx, e);
            return null;
        }
    }
    
    /**
     * 从指定文件系统中读取块数据
     */
    private byte[] readBlockFromFileSystem(String fileSystemName, String fileId, int blkIdx) {
        try {
            // 从ZK元数据中查找文件路径映射
            String metaPath = "/minfs/files";
            if (zkClient.checkExists().forPath(metaPath) == null) {
                return null;
            }
            
            List<String> children = zkClient.getChildren().forPath(metaPath);
            for (String childPath : children) {
                try {
                    byte[] data = zkClient.getData().forPath(metaPath + "/" + childPath);
                    Map<String, Object> meta = fromJsonBytes(data);
                    
                    String metaFileSystem = (String) meta.get("fileSystemName");
                    String filePath = (String) meta.get("path");
                    
                    if (fileSystemName.equals(metaFileSystem) && filePath != null) {
                        // 检查这个文件路径是否对应我们要找的fileId
                        if (isFileMatchingId(fileSystemName, filePath, fileId)) {
                            // 读取文件内容
                            Path actualFile = resolveSafePath(fileSystemName, filePath);
                            if (Files.exists(actualFile)) {
                                // 对于小文件，可能整个文件就是一个块
                                // 这里简化处理，假设blkIdx=0表示整个文件
                                if (blkIdx == 0) {
                                    return Files.readAllBytes(actualFile);
                                }
                                // 对于大文件的块读取，需要根据块大小计算偏移
                                return readFileBlock(actualFile, blkIdx);
                            }
                        }
                    }
                } catch (Exception ignore) {
                    // 继续检查其他文件
                }
            }
            
            return null;
            
        } catch (Exception e) {
            log.debug("Error reading from filesystem {} for {}:{}", fileSystemName, fileId, blkIdx, e);
            return null;
        }
    }
    
    /**
     * 从文件中读取指定块的数据
     */
    private byte[] readFileBlock(Path filePath, int blkIdx) {
        try {
            // 使用与MetaServer相同的块大小（从配置中获取，默认512KB）
            int blockSize = 524288; // 512KB，应该从配置读取
            long offset = (long) blkIdx * blockSize;
            
            long fileSize = Files.size(filePath);
            if (offset >= fileSize) {
                return new byte[0]; // 块超出文件范围
            }
            
            int readSize = (int) Math.min(blockSize, fileSize - offset);
            byte[] buffer = new byte[readSize];
            
            try (RandomAccessFile raf = new RandomAccessFile(filePath.toFile(), "r")) {
                raf.seek(offset);
                int bytesRead = raf.read(buffer);
                if (bytesRead < readSize) {
                    byte[] actualData = new byte[bytesRead];
                    System.arraycopy(buffer, 0, actualData, 0, bytesRead);
                    return actualData;
                }
                return buffer;
            }
            
        } catch (Exception e) {
            log.error("Error reading file block from {}, blkIdx: {}", filePath, blkIdx, e);
            return null;
        }
    }
    
    /**
     * 检查指定的块文件是否存在
     * 支持两种存储模式：块存储和文件存储
     * 
     * @param fileId 文件ID
     * @param blkIdx 块索引
     * @return 块文件是否存在
     */
    public boolean blockExists(String fileId, int blkIdx) {
        try {
            // 首先检查标准块存储路径
            Path blockPath = getBlockPath(fileId, blkIdx);
            boolean blockExists = Files.exists(blockPath);
            
            if (blockExists) {
                log.debug("Block exists check (block storage): {}:{} -> true", fileId, blkIdx);
                return true;
            }
            
            // 如果块存储中不存在，检查是否有文件级存储
            // 尝试从ZK元数据中查找对应的文件路径
            boolean fileExists = checkFileBasedBlock(fileId, blkIdx);
            
            log.debug("Block exists check: {}:{} -> {} (block: {}, file: {})", 
                    fileId, blkIdx, fileExists, blockExists, fileExists);
            return fileExists;
            
        } catch (Exception e) {
            log.error("Error checking block existence for {}:{}", fileId, blkIdx, e);
            return false;
        }
    }
    
    /**
     * 检查文件级存储中是否存在对应的块
     * 通过查找ZK中的文件元数据来定位实际文件
     */
    private boolean checkFileBasedBlock(String fileId, int blkIdx) {
        try {
            // 遍历所有文件系统目录
            Path dataDir = Paths.get(dataPath);
            if (!Files.exists(dataDir)) {
                return false;
            }
            
            try (java.util.stream.Stream<Path> fsDirs = Files.list(dataDir)) {
                return fsDirs.filter(Files::isDirectory)
                        .anyMatch(fsDir -> {
                            String fsName = fsDir.getFileName().toString();
                            if ("blocks".equals(fsName)) return false; // 跳过块存储目录
                            
                            return findFileInFileSystem(fsName, fileId, blkIdx);
                        });
            }
            
        } catch (Exception e) {
            log.debug("Error checking file-based block for {}:{}", fileId, blkIdx, e);
            return false;
        }
    }
    
    /**
     * 在指定文件系统中查找文件
     */
    private boolean findFileInFileSystem(String fileSystemName, String fileId, int blkIdx) {
        try {
            // 从ZK元数据中查找文件路径映射
            String metaPath = "/minfs/files";
            if (zkClient.checkExists().forPath(metaPath) == null) {
                return false;
            }
            
            List<String> children = zkClient.getChildren().forPath(metaPath);
            for (String childPath : children) {
                try {
                    byte[] data = zkClient.getData().forPath(metaPath + "/" + childPath);
                    Map<String, Object> meta = fromJsonBytes(data);
                    
                    String metaFileSystem = (String) meta.get("fileSystemName");
                    String filePath = (String) meta.get("path");
                    
                    if (fileSystemName.equals(metaFileSystem) && filePath != null) {
                        // 检查这个文件路径是否对应我们要找的fileId
                        if (isFileMatchingId(fileSystemName, filePath, fileId)) {
                            // 检查文件是否实际存在
                            Path actualFile = resolveSafePath(fileSystemName, filePath);
                            return Files.exists(actualFile);
                        }
                    }
                } catch (Exception ignore) {
                    // 继续检查其他文件
                }
            }
            
            return false;
            
        } catch (Exception e) {
            log.debug("Error searching in filesystem {} for {}:{}", fileSystemName, fileId, blkIdx, e);
            return false;
        }
    }
    
    /**
     * 检查文件是否匹配指定的fileId
     * 这里实现文件路径到fileId的映射逻辑
     */
    private boolean isFileMatchingId(String fileSystemName, String filePath, String fileId) {
        try {
            // 方案1: 如果fileId就是某种编码的路径，尝试解码
            try {
                String decoded = new String(java.util.Base64.getUrlDecoder().decode(fileId));
                if (decoded.endsWith(filePath) || decoded.contains(filePath)) {
                    return true;
                }
            } catch (Exception ignore) {}
            
            // 方案2: 生成路径的hash并比较
            String pathHash = generateFileId(fileSystemName, filePath);
            if (fileId.equals(pathHash)) {
                return true;
            }
            
            // 方案3: 检查是否fileId直接对应路径（用于调试）
            if (fileId.equals(filePath) || fileId.endsWith(filePath)) {
                return true;
            }
            
            return false;
            
        } catch (Exception e) {
            return false;
        }
    }
    
    /**
     * 根据文件系统名和路径生成fileId（与MetaServer保持一致）
     */
    private String generateFileId(String fileSystemName, String filePath) {
        try {
            String combined = fileSystemName + ":" + filePath;
            java.security.MessageDigest md = java.security.MessageDigest.getInstance("MD5");
            byte[] hash = md.digest(combined.getBytes(java.nio.charset.StandardCharsets.UTF_8));
            StringBuilder sb = new StringBuilder();
            for (byte b : hash) {
                sb.append(String.format("%02x", b));
            }
            return sb.toString();
        } catch (Exception e) {
            return fileSystemName + "_" + filePath.replace("/", "_");
        }
    }

    private byte[] fetchBlockFromRemote(String host, int port, String fileId, int blkIdx, long size) {
        try {
            String url = String.format("http://%s:%d/api/v1/data/block?fileId=%s&blkIdx=%d", host, port, 
                    URLEncoder.encode(fileId, java.nio.charset.StandardCharsets.UTF_8), blkIdx);
            HttpRequest req = HttpRequest.newBuilder()
                    .uri(URI.create(url))
                    .timeout(Duration.ofSeconds(20))
                    .GET().build();
            HttpResponse<byte[]> resp = httpClient.send(req, HttpResponse.BodyHandlers.ofByteArray());
            if (resp.statusCode() == 200) {
                return resp.body();
            }
            return null;
        } catch (Exception e) {
            log.warn("fetchBlockFromRemote failed: {}:{} {}:{}, {}", host, port, fileId, blkIdx, e.getMessage());
            return null;
        }
    }

    private Path writeBlockLocal(String fileId, int blkIdx, byte[] bytes) throws IOException {
        Path p = getBlockPath(fileId, blkIdx);
        Path parent = p.getParent();
        if (parent != null && !Files.exists(parent)) Files.createDirectories(parent);
        Files.write(p, bytes);
        return p;
    }

    private Path getBlockPath(String fileId, int blkIdx) {
        String blocksDirName = "blocks";
        String safeId = sanitizeFileId(fileId);
        Path base = Paths.get(dataPath).resolve(blocksDirName).resolve(safeId);
        String fileName = String.format("blk_%05d.bin", blkIdx);
        return base.resolve(fileName).normalize();
    }

    /**
     * 对 fileId 做简单校验与清理，避免构成异常/危险目录名
     * 规则：
     * - 禁止包含路径分隔符或 ".."
     * - 仅保留 [A-Za-z0-9._-]，其余字符替换为下划线
     * - 去除首尾空白；结果为空则报错
     * - 长度超过 128 时截断
     */
    private String sanitizeFileId(String fileId) {
        if (fileId == null) {
            throw new IllegalArgumentException("fileId is required");
        }
        String trimmed = fileId.trim();
        if (trimmed.isEmpty()) {
            throw new IllegalArgumentException("fileId cannot be empty");
        }
        if (trimmed.contains("../") || trimmed.contains("..\\") || trimmed.contains("..")
                || trimmed.contains("/") || trimmed.contains("\\")) {
            throw new IllegalArgumentException("invalid fileId: contains path separators");
        }
        String cleaned = trimmed.replaceAll("[^A-Za-z0-9._-]", "_");
        if (cleaned.isEmpty()) {
            throw new IllegalArgumentException("invalid fileId after sanitization");
        }
        if (cleaned.length() > 128) {
            cleaned = cleaned.substring(0, 128);
        }
        return cleaned;
    }

    

    private Optional<HostPort> resolveHostPort(String nodeId) {
        try {
            String base = "/minfs/dataservers/" + nodeId;
            byte[] data = zkClient.getData().forPath(base);
            String json = new String(data, StandardCharsets.UTF_8);
            String host = "localhost"; int port = getPortFromNodeId(nodeId);
            try {
                JsonNode j = objectMapper.readTree(json);
                if (j.has("host")) host = j.get("host").asText("localhost");
                if (j.has("port")) port = j.get("port").asInt(port);
            } catch (Exception ignore) {}
            return Optional.of(new HostPort(host, port));
        } catch (Exception e) {
            return Optional.empty();
        }
    }

    private byte[] readAllBytesLocal(String fileSystemName, String path) throws IOException {
        Path filePath = resolveSafePath(fileSystemName, path);
        return Files.readAllBytes(filePath);
    }

    /**
     * 从指定节点获取文件字节
     * @deprecated 文件级存储已废弃，请使用块级存储API
     */
    @Deprecated
    private Optional<byte[]> fetchFileBytesFromNode(String fileSystemName, String path, String sourceNodeId) {
        log.warn("File-level fetchFileBytesFromNode API called but deprecated: fileSystemName={}, path={}, sourceNodeId={}", 
                fileSystemName, path, sourceNodeId);
        return Optional.empty(); // 返回空表示获取失败，符合原有API语义
    }

    private boolean nodeHasFile(String fileSystemName, String path, String node) {
        if (Objects.equals(node, nodeId)) {
            Path p = resolveSafePath(fileSystemName, path);
            return Files.exists(p);
        }
        try {
            Optional<HostPort> hpOpt = resolveHostPort(node);
            if (hpOpt.isEmpty()) return false;
            HostPort hp = hpOpt.get();
            String infoUrl = String.format("http://%s:%d/api/info?path=%s", hp.host, hp.port, path);
            HttpRequest req = HttpRequest.newBuilder()
                    .uri(URI.create(infoUrl))
                    .header("fileSystemName", fileSystemName)
                    .timeout(Duration.ofSeconds(3))
                    .GET().build();
            HttpResponse<String> resp = httpClient.send(req, HttpResponse.BodyHandlers.ofString());
            return resp.statusCode() == 200 && resp.body() != null && resp.body().contains("\"exists\":true");
        } catch (Exception e) {
            return false;
        }
    }

    /**
     * 修复文件
     * @deprecated 文件级存储已废弃，请使用块级存储API
     */
    @Deprecated
    public Map<String, Object> repairFile(String fileSystemName, String path) {
        Map<String, Object> res = new HashMap<>();
        res.put("success", false);
        res.put("message", "File-level repair API is deprecated. Please use block-level storage API");
        res.put("errorCode", "DEPRECATED_API");
        res.put("fileSystemName", fileSystemName);
        res.put("path", path);
        log.warn("File-level repairFile API called but deprecated: fileSystemName={}, path={}", fileSystemName, path);
        return res;
    }

    public Map<String, Object> fsckOnce(int maxFiles) {
        Map<String, Object> summary = new HashMap<>();
        int scanned = 0, repaired = 0, failed = 0, skipped = 0;
        List<Map<String, Object>> details = new ArrayList<>();
        try {
            String base = "/minfs/files";
            if (zkClient.checkExists().forPath(base) == null) {
                summary.put("scanned", 0);
                summary.put("repaired", 0);
                summary.put("failed", 0);
                summary.put("skipped", 0);
                return summary;
            }
            List<String> children = zkClient.getChildren().forPath(base);
            for (String ch : children) {
                if (scanned >= maxFiles) break;
                scanned++;
                byte[] data = zkClient.getData().forPath(base + "/" + ch);
                Map<String, Object> meta = fromJsonBytes(data);
                String fs = meta.getOrDefault("fileSystemName", "").toString();
                String p = meta.getOrDefault("path", "").toString();
                Map<String, Object> r = repairFile(fs, p);
                details.add(r);
                if (Boolean.TRUE.equals(r.get("success"))) {
                    if (Integer.parseInt(String.valueOf(r.getOrDefault("replicaCount", 0))) >= getRequiredReplicas()) repaired++;
                    else skipped++;
                } else {
                    failed++;
                }
            }
        } catch (Exception e) {
            summary.put("error", e.getMessage());
        }
        summary.put("scanned", scanned);
        summary.put("repaired", repaired);
        summary.put("failed", failed);
        summary.put("skipped", skipped);
        summary.put("details", details);
        summary.put("timestamp", System.currentTimeMillis());
        return summary;
    }

    private static final class HostPort {
        final String host; final int port; HostPort(String host, int port){ this.host=host; this.port=port; }
    }

    /**
     * 解析并规范化客户端传入的路径，禁止绝对路径与目录穿越
     */
    private Path resolveSafePath(String fileSystemName, String relativePath) {
        try {
            if (fileSystemName == null || fileSystemName.isBlank()) {
                throw new IllegalArgumentException("fileSystemName is required");
            }
            if (relativePath == null || relativePath.isBlank()) {
                throw new IllegalArgumentException("path is required");
            }
            // 统一去掉开头的分隔符，防止绝对路径覆盖
            String sanitized = relativePath.replace('\\', '/');
            while (sanitized.startsWith("/")) {
                sanitized = sanitized.substring(1);
            }
            Path base = Paths.get(dataPath).resolve(fileSystemName).normalize();
            Path resolved = base.resolve(sanitized).normalize();
            if (!resolved.startsWith(base)) {
                throw new SecurityException("Invalid path: directory traversal detected");
            }
            return resolved;
        } catch (RuntimeException e) {
            throw e;
        }
    }
    
    /**
     * 副本结果内部类
     */
    private static class ReplicaResult {
        final boolean success;
        final String dataServerId;
        final String path;
        final String errorMessage;
        
        ReplicaResult(boolean success, String dataServerId, String path, String errorMessage) {
            this.success = success;
            this.dataServerId = dataServerId;
            this.path = path;
            this.errorMessage = errorMessage;
        }
    }
    
    /**
     * 清理资源
     */
    public void cleanup() {
        if (replicaExecutor != null && !replicaExecutor.isShutdown()) {
            replicaExecutor.shutdown();
            try {
                if (!replicaExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                    replicaExecutor.shutdownNow();
                }
            } catch (InterruptedException e) {
                replicaExecutor.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }
}


